<?php

class CommandProcessor {
	
	private $schedulerService;
	private $taskService;
	private $fileStatSyncConfService;

	public function __construct() {
		$this->schedulerService = SchedulerService::getInstance();
		$this->taskService = TaskService::getInstance();
	}
	
	public function processCommands() {
		Yii::log('begin process commands', CLogger::LEVEL_INFO, SCHEDULER_LOG);
		$commands = $this->schedulerService->getSubmitCommands();
		
		if($commands === false) {
			Yii::log('get commands failed', CLogger::LEVEL_ERROR, SCHEDULER_LOG);
			return;
		}
		
		Yii::log('commands waiting to handle, total: '. count($commands), CLogger::LEVEL_INFO, SCHEDULER_LOG);
		foreach($commands as $command) {
			$command['param'] = CJSON::decode($command['cmd_param']);
			$this->processCommand($command);
		}
		Yii::log('process commands succeed', CLogger::LEVEL_INFO, SCHEDULER_LOG);
	}
	
	public function processCommand($cmd) {
		Yii::log("processing cmd: {$cmd['cmd_type']}", CLogger::LEVEL_INFO, SCHEDULER_LOG);
		switch($cmd['cmd_type']) {
			
			case CommandType::KillTask:
				$ret = $this->killTask($cmd['task_id']);
				break;

			case CommandType::RerunTask:
				$ret = $this->rerunTask($cmd['task_id']);
				break;
				
			case CommandType::ResumeTask:
				$ret = $this->resumeTask($cmd['task_id']);
				break;
				
			case CommandType::RunJob:
				$ret = $this->runJob(
						intval($cmd['param']['entity_id']),
						intval($cmd['param']['type']),
						$cmd['param']['date'],
						$cmd['param']['version'],
						$cmd['param']['priority'],
						isset($cmd['param']['groupid']) ? $cmd['param']['groupid'] : null,
						isset($cmd['param']['is_merged']) ? intval($cmd['param']['is_merged']) : 0,
						$cmd['id']
						);
				break;
				
			default:
				Yii::log("unknown command: {$cmd['cmd']}", CLogger::LEVEL_WARNING, SCHEDULER_LOG);
				$ret = 'unknown command';
				break;
		}

		// command not for me
		if($ret === null) {
			return true;
		}

		$cmdAttributes = array(
			'state' => ($ret === true) ? CmdState::Done : CmdState::Failed,
			'detail'   => ($ret === true) ? 'command execute successfully' : $ret,
			'accept_time' => DateHelper::makeExactDate(),
		);

		while($this->schedulerService->editCommandAttributes($cmd['cmd_id'], $cmdAttributes) === false) {
			Yii::log('update command state failed, keep retrying', CLogger::LEVEL_WARNING, SCHEDULER_LOG);
			sleep(2);
		}

		return true;
	}
	
	/**
	 * Adjust task priority
	 * 
	 * @param int $taskid 
	 * @param int $priority 
	 * @access private
	 * @return bool|string 
	 */
	private function adjustTaskPriority($taskId, $priority) {
		$state = $this->taskService->checkState($taskId, array(
			TaskState::Waiting, 
			TaskState::Ready, 
			TaskState::Pending));
			
		if($state === false) {
			return 'task not found or can not adjust its priority';
		}
			
		if($priority < PriorityConf::MIN || $priority > PriorityConf::MAX) {
			return 'specified priority is not vaild';
		}
		
		$status = $this->taskService->editTaskAttributes($taskId, array('priority'	=>	$priority));
		
		if($status === false) {
			return 'update task priority failed: db error';
		}
		
		return true;
	}

	/**
	 * Only kill task that not in running state
	 * 
	 * @param int $taskid 
	 * @access private
	 * @return bool|string|null return true if kill success, null means ignore, otherwise fail
	 */
	private function killTask($taskId) {

		$state = $this->taskService->checkState($taskId, array(TaskState::Waiting, 
			TaskState::Ready, TaskState::Running, TaskState::Pending));

		if($state === false) {
			return 'task not found or can not kill it';
		}

		if($state == TaskState::Running) {
			// let executor handle this command, ignore it
			return null;
		}
		else {
				
			$status = $this->taskService->editTaskAttributes($taskId, array('state'	=>	TaskState::Killed));
			if($status === false) {
				return 'update task state to killed failed: db error';
			}
			return true;
		}
	}
	
	/**
	 * rerun task 
	 * 
	 * @param int $taskid 
	 * @access private 
	 * @return bool|string
	 */
	private function rerunTask($taskId) {
		
		$state = $this->taskService->checkState($taskId, array(TaskState::Done, TaskState::Failed));
		if($state === false) {
			return "task not found or can not rerun it";
		}
		$task = $this->taskService->getTaskById($taskId);
		// check task duplication
		$duplication = $this->taskService->checkDuplication($task['job_id'], $task['date'], $task['task_type']);		
		if($duplication) {
			return 'one task with the specified job and date is running';
		}
		
		$status = $this->taskService->rerunTask($taskId);
		if($status === false) {
			return 'update task state to ready failed';
		}
		return true;
	}
	
	
	/**
	 * Resume task 
	 * 
	 * @param int $taskid 
	 * @access private 
	 * @return bool|string
	 */
	private function resumeTask($taskId) {
		
		$state = $this->taskService->checkState($taskId, array(TaskState::Failed));
		if($state === false) {
			return "task not found or can not resume it";
		}
		
		$status = $this->taskService->editTaskAttributes($taskId, array(
			'state'	=>	TaskState::Ready,
			'detail'=>	'',
		));
		if($status === false) {
			return 'update task state to ready failed';
		}
		return true;
	}

	/**
	 * Suspend task 
	 * 
	 * @param int $taskid 
	 * @access private 
	 * @return bool|string
	 */
	private function suspendTask($taskId) {
		
		$state = $this->taskService->checkState($taskId, array(TaskState::Waiting, TaskState::Ready));
		if($state === false) {
			return "task not found or can not suspend it";
		}
		
		$status = $this->taskService->editTaskAttributes($taskId, array('state'	=>	TaskState::Pending));
		if($status === false) {
			return 'update task state to pending failed: db error';
		}
		return true;
	}

	/**
	 * recover task
	 * 
	 * @param int $taskid 
	 * @access private
	 * @return bool|string
	 */
	private function recoverTask($taskId) {
	
		$state = $this->taskService->checkState($taskId, array(TaskState::Pending));
		if($state === false) {
			return "task not found or can not resume it";
		}
		
		$status = $this->taskService->editTaskAttributes($taskId, array('state'	=>	TaskState::Waiting));
		if($status === false) {
			return 'update task state to waiting failed: db error';
		}
		return true;
	}

	/**
	 * Run job manually 
	 * 
	 * @param int $entity_id ʵ��id��ͳ��id����־id������id
	 * @param int $type ���ͣ�ͳ����������������
	 * @param string $date ��׼ʱ��
	 * @param int $version ͳ�ư汾
	 * @param int $priority ���ȼ�
	 * @param int $groupid ָ���ķ���id
	 * @param int $cmd_id ���������id
	 * @access private
	 * @return bool|string
	 */
	private function runJob($entityId, $type, $date, $version, $priority, $runResId, $isMerged, $cmdId) {
		
		if($type !== TaskType::Cache && $type !== TaskType::Download
			&& $type != TaskType::Projection && $type != TaskType::Push
			&& $type != TaskType::Statistics) {
			return 'task type error';
		}
		
		// check task duplication
		$duplication = $this->taskService->checkDuplication($entityId, $date, $type);		
		if($duplication) {
			return 'one task with the specified job and date is running';
		}
		
		$jobInfo = $this->dataCache->getJobById($entityId, $type);
		if($jobInfo == null) {
			return 'specified job not exists or the job is not ready';
		}
		
		$task = array();
		$task['entity_id'] = $entityId;
		$task['priority'] = isset($jobInfo['priority']) ? $jobInfo['priority'] : PriorityConf::PDEFAULT;
		$task['pid'] = isset($jobInfo['pid']) ? $jobInfo['pid'] : $jobInfo['product_id'];
		$task['freq'] = isset($jobInfo['freq']) ? $jobInfo['freq'] : $jobInfo['run_cycle'];
		
		if($priority !== null) 	{
			$priority = intval($priority);
			if($priority < PriorityConf::MIN || $priority > PriorityConf::MAX) {
				return 'specified priority is not vaild';
			}
			$task['priority'] = $priority;
		}

		$date = DateHelper::makeDate(strtotime($date));
		if($date <= 0) {
			return 'specified date is not vaild, wrong format';
		}
		else if(strtotime($date) > time()) {
			return 'specified date exceeds the current date';
		}
		
		if($runResId !== null && intval($runResId) > 0 ) {
			$task['runres_id'] = intval($runResId);
		}
		
		if($version !== null && intval($version) >  0) {
			$task['job_version'] = intval($version);
		}

		$task['date'] = $date;
		$task['cmd_id'] = $cmdId;
		$task['type'] = $type;
		$task['state'] = TaskState::Waiting;
		$task['is_merged'] = $isMerged;
		
		$taskId = $this->taskService->addTask($task);
		
		if($taskId === false) {
			return 'db error when inserting manual task';
		}

		return true;
	}

    private function executeCommand($cmd, &$out, $maxRetryTimes = 3) {
        //return Commands::executeCommand($sshCmd, $out);
        Yii::log("execute: $cmd", CLogger::LEVEL_INFO);
        $retryTimes = 0;
        while ($retryTimes++ < $maxRetryTimes) {
            exec($cmd, $out, $ret);
            if ($ret == 0) {
                break;
            }
            sleep($retryTimes * 10);
        }
        return $ret;
    }

    private function remoteExecuteCommand($cmd, $user, $machine, &$out, $retryTimes = 3) {
        $sshCmd = "ssh ".$user."@".$machine . " ";
        $sshCmd .= '"' . $cmd . '"';
        //echo $sshCmd . "\n";
        return $this->executeCommand($sshCmd, $out, $retryTimes);
    }

    private function getkillProcessCommand($name) {
        $cmd = "PID=\`ps axu | grep $name | grep -v 'grep $name' | awk -F ' ' '{print \\\$2}' | head -1\`;";
        $cmd .= "\nif [ -z \"\\\$PID\" ]; then\n\texit 0;\nfi\n";
        $cmd .= "CPIDS=\`pgrep -P \\\$PID\`;"; 
        $cmd .= "\nif [ -n \"\\\$CPIDS\" ]; then kill -TERM \\\$CPIDS;fi;kill \\\$PID;"; 
        return $cmd;
    }

    private function getHadoopJobId($logPath, $user, $machine) {
        $cmd .= "grep 'Running job: ' $logPath | awk -F 'Running job: ' '{print \\\$2}'"; 
        $retval = $this->remoteExecuteCommand($cmd, $user, $machine, $out);
        if ($retval == 0 && count($out) >= 1) {
            return $out[0];
        }
        return false;
    }

    public function cleanTask($task_id, $force = false) {
        if (!$force) {
            $ret = $this->killTask($task_id);
            if ($ret) {
                return true;
            }
        }

        $task = $this->taskService->getTaskById($task_id);
        if (!$task) {
            return false;
        }

        $taskInfo = JsonUtils::jsonDecode($task['runtime']);

        $entity_id = $task['entity_id'];
        $date = $task['date'];
        $clusterId = $task['runResource']['cluster_id'];
        $rt_path = PathUtil::getRuntimePath($task_id, $entity_id, $date);
        $task_dir_name = "${entity_id}_${task_id}";
        $user = LSP_UNAME; 
        $machine = $task['lmachine'];
        $jobType = $taskInfo['basic']['jobType'];
        if ($jobType != JobType::LOCAL_COMPUTE && $jobType != JobType::VISUAL) {
            $logPath = $rt_path . '/log';
            $cmd = "cd $logPath;" . "ls *.log | grep hadoop_.\.log | awk -F '[_|.]' '{if (\\\$2 > max) max=\\\$2} END{print max}'";
            $ret = $this->remoteExecuteCommand($cmd, $user, $machine, $out);
            if ($ret == 0 && count($out) == 1) {
                $step = $out[0];
                $logPath = $rt_path . "/log/hadoop_{$step}.log"; 
                $hadoop_job_id = $this->getHadoopJobId($logPath, $user, $machine);
                if($hadoop_job_id) { 
                    $hadoopClient = PathUtil::getHadoopClientPath($clusterId);
                    $cmd = $hadoopClient ." job -kill $hadoop_job_id;";
                    $ret = $this->remoteExecuteCommand($cmd, $user, $machine, $out, 2);
                } else {
                    echo "WARNING: get hadoop job id failed[$retval]\n";
                }
            }
        }

        $name = "LSP_$task_dir_name";
        $killCmd = $this->getkillProcessCommand($name);

        //ɱ��ִ�л��ϵ�����
        $retval = $this->remoteExecuteCommand($killCmd, $user, $machine, $out);
        if ($retval !== 0) {
            Yii::log("failed to kill process $name at $machine", CLogger::LEVEL_ERROR, EXECUTOR_LOG); 
            $ret = false;
        }

        //ɱ���пػ��ϵ�����
        $name = 'executor/' . $entity_id . '/' . $task_id;
        $killCmd = $this->getkillProcessCommand($name);
        $killCmd = str_replace('\`', '`', $killCmd);
        $killCmd = str_replace('\$', '$', $killCmd);
        //echo $killCmd;
        $retval = $this->executeCommand($killCmd, $out);
        if ($retval !== 0) {
            Yii::log("failed to kill process $name", CLogger::LEVEL_ERROR, EXECUTOR_LOG); 
            $ret = false;
        }

        //�����ݿ��е�״̬
        $row = array('detail'=>'task cleaned', 'state'=>TaskState::Killed);
        $retryTimes = 0;
        while ($retryTimes++ < 3) {
            if ($this->taskService->editTaskAttributes($task_id, $row) !== false) {
                break;
            } 
            sleep(5 * $retryTimes);
        }
        if ($retryTimes > 3) {
            Yii::log("failed to update task's state to killed", CLogger::LEVEL_ERROR, EXECUTOR_LOG);
            $ret = false;
        }
        //������ݿ������״̬
        $param = array('type'=>$task['type'], 'entity_id'=>$entity_id, 'id'=>$task_id);
        $row = array('state'=>CmdState::Done, 'accept_time'=>DateHelper::makeExactDate(), 'param'=>serialize($param));
        if (!$ret) {
            $row['state'] = CmdState::Failed;
            $row['ret'] = "failed to clean task";
        } 
        $retryTimes = 0;
        while ($retryTimes++ < 3) {
            if ($this->schedulerService->editCommandAttributes($cmd_id, $row) !== false) {
                break;
            } 
            sleep(5 * $retryTimes);
        }
        if ($retryTimes > 3) {
            Yii::log("failed to update command, state", CLogger::LEVEL_WARNING, EXECUTOR_LOG);
            return false; 
        }
        return true; 
    }
}
